﻿#include "net/mqtt/acl_mqtt_server_adapter.h"
# if acl
mqtt_async_client::mqtt_async_client(std::string mClientId, bool cleanSession) : m_client_id_(std::move(mClientId)),
                                                                                 clean_session_(cleanSession) {}

void mqtt_async_client::run() {
    boost::asio::io_context ioc;

    // Create no TLS client
    auto c = MQTT_NS::make_async_client(ioc, ip, port);

    // Setup client
    c->set_client_id("cid1");
    c->set_clean_session(true);

    boost::asio::steady_timer tim_connect_wait_limit(ioc);
    boost::asio::steady_timer tim_reconnect_delay(ioc);

    std::function < void() > connect;
    std::function < void() > reconnect;

    connect =
            [&]
            {
                LOG_INFO << "async_connect";
                c->async_connect(
                        [&]
                                (MQTT_NS::error_code ec)
                        {
                            LOG_INFO << "async_connect callback: " << ec.message();
                            if (ec) reconnect();
                        }
                );

                LOG_INFO << "tim_connect_wait_limit set";
                tim_connect_wait_limit.expires_after(std::chrono::seconds(3));
                tim_connect_wait_limit.async_wait(
                        [&](boost::system::error_code ec)
                        {
                            LOG_INFO << "tim_connect_wait_limit callback: " << ec.message();
                            if (!ec)
                            {
                                c->async_force_disconnect(
                                        [&](boost::system::error_code ec)
                                        {
                                            LOG_INFO << "async_force_disconnect callback: " << ec.message();
                                        }
                                );
                            }
                        }
                );
            };

    reconnect =
            [&]
            {
                LOG_INFO << "tim_reconnect_delay set";
                tim_reconnect_delay.expires_after(std::chrono::seconds(3));
                tim_reconnect_delay.async_wait(
                        [&](boost::system::error_code ec)
                        {
                            LOG_INFO << "tim_reconnect_delay callback: " << ec.message();
                            if (!ec) connect();
                        }
                );
            };

    // Setup handlers
    c->set_connack_handler(
            [&]
                    (bool sp, MQTT_NS::connect_return_code connack_return_code)
            {
                LOG_INFO << "Connack handler called";
                LOG_INFO << "  Session Present: " << std::boolalpha << sp;
                LOG_INFO << "  Connack Return Code: "
                         << MQTT_NS::connect_return_code_to_str(connack_return_code);
                if (connack_return_code == MQTT_NS::connect_return_code::accepted)
                {
                    tim_connect_wait_limit.cancel();
                }
                else
                {
                    reconnect();
                }
                return true;
            });
    c->set_close_handler(
            [&]
                    ()
            {
                LOG_INFO << "closed.";
                reconnect();
            });
    c->set_error_handler(
            [&]
                    (MQTT_NS::error_code ec)
            {
                LOG_INFO << "error: " << ec.message();
                reconnect();
            });

    connect();
    ioc.run();
}

template<typename Client, typename Disconnect>
void client_proc(
        Client &c,
        std::uint16_t &pid_sub1,
        std::uint16_t &pid_sub2,
        Disconnect const &disconnect) {

    using packet_id_t = typename std::remove_reference_t<decltype(*c)>::packet_id_t;
    // Setup client
    c->set_client_id("cid1");
    c->set_clean_session(true);

    // Setup handlers
    c->set_connack_handler(
            [&c, &pid_sub1, &pid_sub2]
                    (bool sp, MQTT_NS::connect_return_code connack_return_code)
            {
                LOG_INFO << "[client] Connack handler called";
                LOG_INFO << "[client] Session Present: " << std::boolalpha << sp;
                LOG_INFO << "[client] Connack Return Code: "
                         << MQTT_NS::connect_return_code_to_str(connack_return_code);
                if (connack_return_code == MQTT_NS::connect_return_code::accepted)
                {
                    pid_sub1 = c->subscribe("mqtt_client_cpp/topic1", MQTT_NS::qos::at_most_once);
                    pid_sub2 = c->subscribe(
                            std::vector<std::tuple<MQTT_NS::string_view, MQTT_NS::subscribe_options>>
                                    {
                                            {"mqtt_client_cpp/topic2_1", MQTT_NS::qos::at_least_once},
                                            {"mqtt_client_cpp/topic2_2", MQTT_NS::qos::exactly_once}
                                    }
                    );
                }
                return true;
            });
    c->set_close_handler(
            []()
            {
                LOG_INFO << "[client] closed.";
            });
    c->set_error_handler(
            [](MQTT_NS::error_code ec)
            {
                LOG_INFO << "[client] error: " << ec.message();
            });
    c->set_puback_handler(
            [&](packet_id_t packet_id)
            {
                LOG_INFO << "[client] puback received. packet_id: " << packet_id;
                disconnect();
                return true;
            });
    c->set_pubrec_handler(
            [&](packet_id_t packet_id)
            {
                LOG_INFO << "[client] pubrec received. packet_id: " << packet_id;
                return true;
            });
    c->set_pubcomp_handler(
            [&](packet_id_t packet_id)
            {
                LOG_INFO << "[client] pubcomp received. packet_id: " << packet_id;
                disconnect();
                return true;
            });
    c->set_suback_handler(
            [&](packet_id_t packet_id, std::vector<MQTT_NS::suback_return_code> results)
            {
                LOG_INFO << "[client] suback received. packet_id: " << packet_id;
                for (auto const &e: results)
                {
                    LOG_INFO << "[client] subscribe result: " << e;
                }
                if (packet_id == pid_sub1)
                {
                    c->publish("mqtt_client_cpp/topic1", "test1", MQTT_NS::qos::at_most_once);
                }
                else if (packet_id == pid_sub2)
                {
                    c->publish("mqtt_client_cpp/topic2_1", "test2_1", MQTT_NS::qos::at_least_once);
                    c->publish("mqtt_client_cpp/topic2_2", "test2_2", MQTT_NS::qos::exactly_once);
                }
                return true;
            });
    c->set_publish_handler(
            [&](MQTT_NS::optional<packet_id_t> packet_id,
                MQTT_NS::publish_options pubopts,
                MQTT_NS::buffer topic_name,
                MQTT_NS::buffer contents)
            {
                LOG_INFO << "[client] publish received. "
                         << " dup: " << pubopts.get_dup()
                         << " qos: " << pubopts.get_qos()
                         << " retain: " << pubopts.get_retain();
                if (packet_id)
                    LOG_INFO << "[client] packet_id: " << *packet_id;
                LOG_INFO << "[client] topic_name: " << topic_name;
                LOG_INFO << "[client] contents: " << contents;
                disconnect();
                return true;
            });

    // Connect
    c->connect();
}



//////////////////////////////////////////////////////////////////////////////

bool mqtt_clients::add(mqtt_client *client) {
    auto cit = clients_.find(client);
    if (cit == clients_.end())
    {
        clients_.insert(client);
        printf("topic=%s, client=%p add ok\r\n",
               topic_.c_str(), client);
        return true;
    }
    else
    {
        printf("topic=%s, client=%p, add error, exist\r\n",
               topic_.c_str(), client);
        return false;
    }
}

bool mqtt_clients::del(mqtt_client *client) {
    auto it = clients_.find(client);
    if (it != clients_.end())
    {
        clients_.erase(it);
        printf("topic=%s, client=%p, del ok\r\n",
               topic_.c_str(), client);
        return true;
    }
    printf("topic=%s, client=%p, del error, not exit\r\n",
           topic_.c_str(), client);
    return false;
}

void mqtt_clients::clear_all() {
    for (auto it = clients_.begin();
         it != clients_.end(); ++it)
    {
        (*it)->close();
    }

    clients_.clear();
}

//////////////////////////////////////////////////////////////////////////////

mqtt_manager::~mqtt_manager(void) {
    for (auto it = manager_.begin(); it != manager_.end(); ++it)
    {
        it->second->clear_all();
        delete it->second;
    }
}

bool mqtt_manager::add(const char *topic, mqtt_client *client) {
    auto it = manager_.find(topic);
    if (it == manager_.end())
    {
        mqtt_clients *clients = new mqtt_clients(topic);
        manager_[topic] = clients;
        return clients->add(client);
    }
    else
    {
        return it->second->add(client);
    }
}

bool mqtt_manager::del(mqtt_client *client) {
    const std::set<acl::string> &topics = client->get_topics();
    if (topics.empty())
    {
        printf("no topics in client=%p\r\n", client);
        return false;
    }

    for (auto cit = topics.begin(); cit != topics.end(); ++cit)
    {
        del(*cit, client);
    }
    return true;
}


void mqtt_manager::del(const char *topic, mqtt_client *client) {
    auto it = manager_.find(topic);
    if (it != manager_.end())
    {
        it->second->del(client);
        if (it->second->empty())
        {
            printf("topic=%s, delete mqtt_clients\r\n",
                   it->second->get_topic());
            delete it->second;
            manager_.erase(it);
        }
    }
}

//////////////////////////////////////////////////////////////////////////////

mqtt_client::mqtt_client(acl::aio_handle &handle, mqtt_manager &manager)
        : acl::mqtt_aclient(handle), manager_(manager) {}

mqtt_client::~mqtt_client() {}

bool mqtt_client::on_open() {
    printf("one client connected\r\n");
    return true;
}

bool mqtt_client::on_header(const acl::mqtt_header &header) {
    printf("got new mqtt header, type=%s\r\n",
           acl::mqtt_type_desc(header.get_type()));
    return true;
}

bool mqtt_client::on_body(const acl::mqtt_message &body) {
    acl::mqtt_type_t type = body.get_header().get_type();
    switch (type)
    {
        case acl::MQTT_CONNECT:
            return handle_connect(body);
        case acl::MQTT_SUBSCRIBE:
            return handle_subscribe(body);
        case acl::MQTT_PINGREQ:
            return handle_pingreq(body);
        case acl::MQTT_PINGRESP:
            return handle_pingresp(body);
        case acl::MQTT_DISCONNECT:
            return handle_disconnect(body);
        case acl::MQTT_PUBLISH:
            return handle_publish(body);
        case acl::MQTT_PUBACK:
            return handle_puback(body);
        default:
            printf("unknown type=%d, %s\r\n",
                   (int) type, acl::mqtt_type_desc(type));
            return true;
    }
}

bool mqtt_client::handle_connect(const acl::mqtt_message &body) {
    // 连接返回代码
    acl::mqtt_connack connack;


    const auto &connect = (const acl::mqtt_connect &) body;

    LOG_INFO << "get_username() :" << (connect.get_username() == NULL ? "null" : connect.get_username()) << ", "
             << "get_passwd()" << (connect.get_passwd() == NULL ? "null" : connect.get_passwd()) << " , "
             << "get_cid() :" << (connect.get_cid() == NULL ? "null" : connect.get_cid()) << " , "
             << "get_will_qos() :" << (connect.get_will_qos() == NULL ? 0x0 : connect.get_will_qos()) << " , "
             << "get_will_topic():" << (connect.get_will_topic() == NULL ? "null" : connect.get_will_topic()) << " , "
             << "get_will_msg() :" << (connect.get_will_msg() == NULL ? "null" : connect.get_will_msg());


    std::string admin = "admin";

    if (connect.get_username() != NULL && admin ==connect.get_username())
    {
        connack.set_connack_code(acl::MQTT_CONNACK_OK);
    }
    else
    {
        connack.set_connack_code(acl::MQTT_CONNACK_ERR_AUTH);
    }

    return this->send(connack);
}

bool mqtt_client::handle_subscribe(const acl::mqtt_message &body) {
    const acl::mqtt_subscribe &subscribe = (const acl::mqtt_subscribe &) body;
    const std::vector<acl::string> &topics = subscribe.get_topics();
    const std::vector<acl::mqtt_qos_t> &qoses = subscribe.get_qoses();

    if (topics.empty())
    {
        printf("not topic got\r\n");
        return false;
    }
    if (qoses.size() != topics.size())
    {
        printf("qos count(%zd) != topic count(%zd)\r\n",
               qoses.size(), topics.size());
        return true;
    }

    LOG_INFO << "pkt_id = " << subscribe.get_pkt_id();
    size_t n = topics.size();

    for (size_t i = 0; i < n; i++)
    {
        LOG_INFO << "topic=" << topics[i].c_str() << "  , qos=" << acl::mqtt_qos_desc(qoses[i]);
        topics_.insert(topics[i]);

        if (!manager_.add(topics[i], this))
        {
            LOG_INFO << "add to manager error" << this;
            return false;
        }
    }

    acl::mqtt_suback suback;
    suback.set_pkt_id(subscribe.get_pkt_id());
    suback.add_topic_qos(qoses);

    LOG_INFO << "  send suback ";
    return this->send(suback);
}

bool mqtt_client::handle_pingreq(const acl::mqtt_message &) {
    acl::mqtt_pingresp pingresp;
    if (this->send(pingresp))
    {
        return true;
    }

    LOG_INFO << "  send pingresp error ";
    return false;
}

bool mqtt_client::handle_pingresp(const acl::mqtt_message &) {
    LOG_INFO << "  got pingresp ";
    return true;
}

bool mqtt_client::handle_disconnect(const acl::mqtt_message &) {
    LOG_INFO << "disconnect ";
    return false;
}

void mqtt_client::forward_publish(const acl::mqtt_publish &publish) {
    std::set<mqtt_client *> *clients =
            manager_.get_clients(publish.get_topic());
    if (clients == NULL || clients->empty())
    {
        return;
    }

    for (std::set<mqtt_client *>::iterator it = clients->begin();
         it != clients->end(); ++it)
    {
        forward_publish(**it, publish);
    }
}

void mqtt_client::forward_publish(mqtt_client &client,
                                  const acl::mqtt_publish &in) {

    acl::mqtt_publish publish;
    publish.get_header().set_qos(in.get_header().get_qos());
    publish.set_pkt_id(in.get_pkt_id());
    publish.set_topic(in.get_topic());
    const acl::string &payload = in.get_payload();
    publish.set_payload((unsigned) payload.size(), payload);
    if (client.send(publish))
    {
        LOG_INFO << "send publish ok ";
    }
    else
    {
        LOG_ERROR << "send publish error ";
    }
}

bool mqtt_client::handle_publish(const acl::mqtt_message &body) {
    const acl::mqtt_publish &publish = (const acl::mqtt_publish &) body;
    const acl::string &payload = publish.get_payload();
    const char *topic = publish.get_topic();
    LOG_DEBUG << "topic: " << topic << " , qos: " << publish.get_header().get_qos()
              << "  ,pkt_id: " << publish.get_pkt_id() << " payload: \r\n" << payload.c_str();
    forward_publish(publish);

    if (publish.get_header().get_qos() == acl::MQTT_QOS0)
    {
        return true;
    }

    acl::mqtt_qos_t qos = body.get_header().get_qos();
    if (qos != acl::MQTT_QOS1)
    {
        printf("topic=%s, qos=%s, needn't ack\r\n",
               publish.get_topic(), acl::mqtt_qos_desc(qos));
        return true;
    }

    acl::mqtt_puback puback;
    puback.set_pkt_id(publish.get_pkt_id());
    if (!this->send(puback))
    {

        LOG_ERROR << "puback error, pkt_id = " << puback.get_pkt_id();
        return false;
    }
    LOG_INFO << "puback ok, pkt_id = " << puback.get_pkt_id();
    return true;
}

bool mqtt_client::handle_puback(const acl::mqtt_message &) {
    return true;
}
#endif